package com.pptv.ads.client.connector;


import java.io.IOException;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;

import com.pptv.ads.client.message.MessageHandler;
import com.pptv.ads.core.Connector;
import com.pptv.ads.core.model.Message;
import com.pptv.ads.core.utils.CuratorZKUtil;
import com.pptv.ads.core.utils.SerializeUtil;

import static com.pptv.ads.core.utils.PathUtil.*;

public class ZKClientConnector implements Connector{

	private String rootPath;
	private String listenPath;
	private CuratorFramework client;
	private MessageHandler handler ;
	
	public ZKClientConnector(CuratorFramework client, String rootPath,String listenPath) {
		this.rootPath = rootPath;
		this.client = client;
		this.listenPath = listenPath;
		handler = new MessageHandler();
	}
	
/*	public void send(Job job) {
		Message msg = new Message(MessageType.JOB, SerializeUtil.serialize(job));

		String jobPath= combinePath(rootPath, ZK_EVENT_PATH + "/"+generateEventId());
		zkOperator.createPersistent(jobPath, SerializeUtil.serialize(msg));
	}*/

	public void accept() {
		PathChildrenCache msgCache = new PathChildrenCache(client,listenPath,true);
		msgCache.getListenable().addListener(new PathChildrenCacheListener() {
			public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
				switch (event.getType()) {
				case CHILD_REMOVED:
					break;
				case CHILD_ADDED:
					ChildData child = event.getData();
//					String s = new String(child.getData());
					Message msg = SerializeUtil.deserialize(child.getData(), Message.class);
					//gson.fromJson(s,new TypeToken<Message<Job>>(){}.getType());
//					Message<Job> msg = SerializeUtil.deserialize(child.getData(), Message.class);
					handler.handler(msg);
					
					CuratorZKUtil.delete(client, child.getPath());
					break;

				default:
					break;
				}
			}

		});
		try {
			msgCache.start();
		} catch (Exception e1) {
			e1.printStackTrace();
			try {
				msgCache.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		

	}

	public void send(Message message) {
		String jobPath= combinePath(rootPath, ZK_EVENT_PATH + "/"+message.getId());
		CuratorZKUtil.createPersistent(client, jobPath, SerializeUtil.serialize(message));
	}

	public void destory() {
		// TODO Auto-generated method stub
		
	}


}
